package org.lwd.datalink.service.impl;

import com.google.common.collect.Lists;
import lombok.extern.slf4j.Slf4j;
import org.lwd.datalink.domain.task.ActiveTasks;
import org.lwd.datalink.domain.task.TaskInfo;
import org.lwd.datalink.errors.DatalinkException;
import org.lwd.datalink.mapper.TaskMapper;
import org.lwd.datalink.service.TaskConfigService;
import org.springframework.stereotype.Service;
import org.springframework.util.Assert;

import javax.annotation.Resource;
import java.util.Date;
import java.util.List;
import java.util.stream.Collectors;

/**
 * 同步任务配置 Service 接口实现
 *
 * @Author: weidong.lin
 * @Date: create in 2024/4/2 14:08
 */
@Slf4j
@Service
public class TaskConfigServiceImpl implements TaskConfigService {

    @Resource
    private TaskMapper taskMapper;

    @Override
    public TaskInfo getTask(Long id) {
        return taskMapper.findById(id);
    }

    @Override
    public ActiveTasks getActiveTaskConfigsByGroup(Long groupId) {
        Assert.notNull(groupId);

        while (true) {
            long start = System.currentTimeMillis();
            long version = getTaskConfigVersion();
            List<TaskInfo> taskInfos = taskMapper.listByGroupId(groupId);
            log.info("Time for querying task from db  is : " + (System.currentTimeMillis() - start) + "ms");

            List<TaskInfo> result;
            if (version == -1 || taskInfos == null) {
                return new ActiveTasks(groupId, version, Lists.newArrayList());
            } else {
                result = taskInfos.stream().filter(t -> t.getUpdateTime().getTime() <= version).collect(Collectors.toList());
                result.forEach(t -> t.setVersion(version));

                if (result.size() == taskInfos.size()) {
                    return new ActiveTasks(groupId, version, result);
                } else {
                    log.info("find dirty tasks which exceed version {}.", version);
                }
            }
        }
    }

    @Override
    public Long getTaskConfigVersion() {
        for (int i = 0; i < 3; i++) {
            try {
                Date maxModifyTime = taskMapper.findMaxModifyTime();
                return maxModifyTime == null ? -1L : maxModifyTime.getTime();
            } catch (Throwable t) {
                // 1.连接闪断的情况下存在异常的可能，需要通过重试进行容错处理
                // 2.DB宕机的情况下，也不要向外抛异常，因为DB宕机期间只要不触发rebalance，系统是可以继续运行的
                //TODO,增加报警功能
                log.error("Get Max ModifyTime Failed!", t);
                try {
                    Thread.sleep(100L);
                } catch (Exception e) {
                }
            }
        }
        throw new DatalinkException("get task config version failed after 3 times.");
    }
}
